/**
 * Krishna Gajula
 * 
 * This class finds and generate report on CDH4 backward compatibility issues with in pig, pig UDFs, Hive, FSshell and MR files.
 */
package com.searshc.hadoop.migration;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.filefilter.RegexFileFilter;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.searshc.hadoop.migration.util.PDFFileGenerator;
import com.searshc.hadoop.migration.util.PatternConstants;

public class CDH4BackwardCompatibilityIssueFinder {
	private static Logger logger = Logger.getLogger("CDH4BackwardCompatibilityIssueFinder");	
	private PDFFileGenerator pdfFileGenerator;
	public CDH4BackwardCompatibilityIssueFinder(){
		ApplicationContext context = new ClassPathXmlApplicationContext(
				"cdh4-migration-spring-context.xml");
		pdfFileGenerator=(PDFFileGenerator)context.getBean("pdfFileGenerator");
		
	}
	public void findIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap){
		Integer scriptExtentionNum=PatternConstants.scripExtentioNumMap.get(scriptExtension);		
        switch (scriptExtentionNum) {
            case 1:  
            	findPigScriptIssues(scriptExtension, directoryPath, fileIssueMap);
            	break;
            case 2:  
            	findHiveScriptIssues(scriptExtension, directoryPath, fileIssueMap);
            	break;
            case 3:  
            	findHDFSCommandIssues(scriptExtension, directoryPath, fileIssueMap);
            	break;
            case 4: 
            	findMapReduceIssues(scriptExtension, directoryPath, fileIssueMap);
            	break;          
            default: 
            	findPigScriptIssues(scriptExtension, directoryPath, fileIssueMap);
            	break;
        }		
	}	
	/**
	 * A common method to find all kinds of compatibility issues
	 * @param scriptExtension
	 * @param directoryPath
	 * @param fileIssueMap
	 */
	public void findScriptIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap, FileSystem fs){
//		Map<String, String> issueMsgMap = PatternConstants.scripExtentionIssuePatternsMap.get(scriptExtension);
		Collection<File> pigScriptFiles = findFiles(scriptExtension, directoryPath);
		
		for(File pigFile:pigScriptFiles){
			Map<String, String> issueLineErrorMsgMap = new HashMap<String, String>();
			try {
				List<String> lines = FileUtils.readLines(pigFile);
				int i=0;
				for(String line:lines){
					populateIssueMsg(line, ++i, scriptExtension,issueLineErrorMsgMap);
					if(fs!=null){
						populatePathIssueMsg(line, i, scriptExtension, issueLineErrorMsgMap, fs);
					}
				}
			} catch (IOException e) {
				logger.error("Can't read file", e);
			}
			fileIssueMap.put(pigFile.getAbsolutePath(), issueLineErrorMsgMap);
		}		
	}
	/**
	 *  Generates pdf report
	 * @param fileIssueMap
	 *
	 */
	public void generateIssueReport(Map<String, Map<String,String>> fileIssueMap){
		pdfFileGenerator.createFile(fileIssueMap);
	}
	/**
	 * Wrapper methods for specific file type issues.
	 * @param scriptExtension
	 * @param directoryPath
	 * @param fileIssueMap
	 */
	public void findPigScriptIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap){
		findScriptIssues(scriptExtension, directoryPath, fileIssueMap, null);		
	}
	public void findHiveScriptIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap){
		findScriptIssues(scriptExtension, directoryPath, fileIssueMap, null);
	}
	public void findMapReduceIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap){
		findScriptIssues(scriptExtension, directoryPath, fileIssueMap, null);
	}
	public void findHDFSCommandIssues(String scriptExtension, String directoryPath, Map<String, Map<String,String>> fileIssueMap){		
		Configuration conf = new Configuration();
        conf.set("fs.defaultFS", PatternConstants.NAMENODE); 
        FileSystem fs = null;
        try {
        	fs = FileSystem.get(conf);
		} catch (IOException e) {
			logger.error("Can't find hdfs", e);
		}
        findScriptIssues(scriptExtension, directoryPath, fileIssueMap, fs);
	}
	/**
	 * finds all the files in a directory recursively by a file extension
	 * @param scriptExtension
	 * @param directoryPath
	 * @return
	 */
	public Collection<File> findFiles(String scriptExtension, String directoryPath){
		File directory = new File(directoryPath);
//		File[] files = directory.listFiles();		
		String pattern = "^.+\\."+scriptExtension+"$";
		logger.info("\nFiles that match regular expression: " + pattern);
//		FileFilter filter = new RegexFileFilter(pattern);
		return FileUtils.listFiles(directory, FileFilterUtils.suffixFileFilter(scriptExtension), TrueFileFilter.INSTANCE);		
	}
	public void populateIssueMsg(String line, Integer lineNum, String scriptExtension, Map<String, String> issueLineErrorMsgMap){		
		Map<String, String> issueMsgMap = PatternConstants.scripExtentionIssuePatternsMap.get(scriptExtension);
		for(String pattern:issueMsgMap.keySet()){
			Pattern ptn = Pattern.compile(pattern, Pattern.CASE_INSENSITIVE);         
			Matcher mtch = ptn.matcher(line);         
			if(mtch.find()){             
				issueLineErrorMsgMap.put(lineNum.toString()+"|"+mtch.group(), issueMsgMap.get(pattern)) ; 				
			}			 
		}		
	} 
	public void populatePathIssueMsg(String line, Integer lineNum, String scriptExtension, Map<String, String> issueLineErrorMsgMap, FileSystem fs){		
//		Map<String, String> issueMsgMap = PatternConstants.shellScriptPathIssuePatternErrorMsgMap.get(scriptExtension);
		for(String pattern:PatternConstants.shellScriptPathIssuePatternErrorMsgMap.keySet()){
			Pattern ptn = Pattern.compile(pattern, Pattern.CASE_INSENSITIVE);         
			Matcher mtch = ptn.matcher(line);         
			if(mtch.find()){ 
				Integer captureGrp = PatternConstants.shellScriptPathIssuePatternErrorMsgMap.get(pattern);
				String pathStr = mtch.group(captureGrp);
				String[] paths = pathStr.split("\\s+");
				for(String path:paths){
					if(!(path.trim().startsWith("/")|| path.trim().toLowerCase().startsWith("hdfs"))){
						issueLineErrorMsgMap.put(lineNum.toString()+"|"+path, "Relative paths will no longer be converted to an absolute path, nor to a fully qualified URI. Replace it with absolute path.") ; 
					}
					try {
						if(!((new File(path)).exists()||(fs.isDirectory(new Path(path)) && fs.exists(new Path(path))))){
							issueLineErrorMsgMap.put(lineNum.toString()+"|"+path, "fs shell no longer auto create non existing directories. You need to create "+path+" manually.") ;
						}
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}								
			}			 
		}		
	}    
	public static void main(String a[]){    
		if(a==null || a.length<2 || !(new File(a[0])).isDirectory()){
			return;
		}else{
			Map<String, Map<String,String>> fileIssueMap = new HashMap<String, Map<String,String>>();
			CDH4BackwardCompatibilityIssueFinder finder = new CDH4BackwardCompatibilityIssueFinder();   
			for(String fileExt:a){
				if(PatternConstants.scripExtentioNumMap.keySet().contains(fileExt.toLowerCase())){					 
					finder.findIssues(fileExt.toLowerCase(), a[0], fileIssueMap);
				}
			}
			if(!fileIssueMap.isEmpty()){
				finder.generateIssueReport(fileIssueMap);
			}
//			CDH4BackwardCompatibilityIssueFinder finder = new CDH4BackwardCompatibilityIssueFinder();         
//			finder.findIssues("pig", "C:\\Sears_projects\\pig_scripts");
	//		finder.findIssues("java", "C:\\Sears_projects\\hive_queries");
		}
	}

}
